package com.example.rocketmq.demo.producer;

import com.example.rocketmq.demo.config.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 事务消息生产者
 */
@Slf4j
@Component
public class TransactionProducer {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    private TransactionMQProducer producer;
    private ExecutorService executorService;

    @PostConstruct
    public void init() {
        // 创建事务消息生产者
        producer = new TransactionMQProducer("transaction-producer-group");
        producer.setNamesrvAddr(nameServer);

        // 创建线程池
        executorService = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2000),
                new ThreadFactory() {
                    private final AtomicInteger threadIndex = new AtomicInteger(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("transaction-msg-check-thread-" + threadIndex.incrementAndGet());
                        return thread;
                    }
                }
        );

        producer.setExecutorService(executorService);

        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                String transactionId = msg.getTransactionId();
                String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                log.info("执行本地事务, transactionId={}, content={}", transactionId, content);

                try {
                    // 模拟本地事务执行
                    String status = (String) arg;
                    if ("COMMIT".equals(status)) {
                        // 本地事务成功，提交事务消息
                        log.info("本地事务执行成功，提交事务消息");
                        return LocalTransactionState.COMMIT_MESSAGE;
                    } else if ("ROLLBACK".equals(status)) {
                        // 本地事务失败，回滚事务消息
                        log.info("本地事务执行失败，回滚事务消息");
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    } else {
                        // 本地事务未知状态，等待回查
                        log.info("本地事务执行未知，等待回查");
                        return LocalTransactionState.UNKNOW;
                    }
                } catch (Exception e) {
                    log.error("本地事务执行异常", e);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String transactionId = msg.getTransactionId();
                String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                log.info("回查本地事务状态, transactionId={}, content={}", transactionId, content);

                // 模拟回查本地事务，这里随机返回结果
                // 实际应用中应该根据业务逻辑查询本地事务执行结果
                int status = ThreadLocalRandom.current().nextInt(3);
                switch (status) {
                    case 0:
                        log.info("回查结果: 提交事务消息");
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case 1:
                        log.info("回查结果: 回滚事务消息");
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    case 2:
                        log.info("回查结果: 继续等待");
                        return LocalTransactionState.UNKNOW;
                    default:
                        return LocalTransactionState.COMMIT_MESSAGE;
                }
            }
        });

        try {
            producer.start();
            log.info("事务消息生产者启动成功");
        } catch (Exception e) {
            log.error("事务消息生产者启动失败", e);
        }
    }

    /**
     * 发送事务消息
     *
     * @param content 消息内容
     * @param status 事务状态 (COMMIT/ROLLBACK/UNKNOWN)
     * @return 发送结果
     */
    public TransactionSendResult sendTransaction(String content, String status) throws Exception {
        Message message = new Message(
                RocketMQConstant.TOPIC_TRANSACTION,
                RocketMQConstant.TAG_TRANSACTION,
                content.getBytes(StandardCharsets.UTF_8)
        );

        TransactionSendResult sendResult = producer.sendMessageInTransaction(message, status);
        log.info("事务消息发送结果: {}", sendResult);
        return sendResult;
    }

    @PreDestroy
    public void destroy() {
        if (producer != null) {
            producer.shutdown();
            log.info("事务消息生产者已关闭");
        }

        if (executorService != null) {
            executorService.shutdown();
            log.info("事务消息线程池已关闭");
        }
    }
}
